/*
 * File Name:msg_bus_test.c
 * Version:v1.0.0
 * Change Logs:
 * Date             Author      	main change
 * 2025-03-07      	lanxin          the first version
 *
*/

/********************************************* READ ME BEGIN *************************************************************/

/* Brief:
 *
 * *1.对 msg_bus 的测试程序。
 *
 */

/* Notice:
 *
 * *
 *
 */

/********************************************* READ ME END ***************************************************************/

/* source file */
#include "msg_bus.h"

/* debug define */

/* private define */
#define TOPIC_MSG_SIZE sizeof(s_topic_msg_t)

/* function declaration */

/* variable declaration */

typedef struct msg
{
	char p_buf[64];
	uint16_t len;
} s_topic_msg_t;

static char *topic_1 = "topic_1";
static char *topic_2 = "topic_2";
static char *topic_3 = "topic_3";
static rt_mq_t mq1, mq2, mq3;

//do_topic_enable topoic_1 enable
static int do_topic_enable ( int argc, char **argv )
{
	if ( argc == 3 )
	{
        //not "enable" set to disable 
		int enable = ( rt_strcmp ( "enable", argv[2] ) == 0 );
		msg_bus_set_topic_enable ( argv[1], enable );
	}

	return 0;
}

static void recv_cb1 ( void *para )
{
	rt_kprintf ( "recv cb:%s\r\n", ( char * ) para );
}

static void recv_cb2 ( void *para )
{
	rt_kprintf ( "recv cb:%s\r\n", ( char * ) para );
}

static int do_topic_unsub ( void )
{
	msg_bus_topic_unsubscribe ( topic_1, mq1 );
	msg_bus_topic_unsubscribe_function ( topic_1, recv_cb1 );
	return 0;
}

static int do_topic_sub ( void )
{
	msg_bus_topic_subscribe ( topic_1, mq1 );
	msg_bus_topic_subscribe_function ( topic_1, recv_cb1, "topic 1 cb" );
	return 0;
}

static void recv_1_entry ( void *para )
{
	mq1 = rt_mq_create ( "mq1", sizeof ( s_topic_msg_t ), 16, RT_IPC_FLAG_FIFO );
	RT_ASSERT ( mq1 );

	msg_bus_topic_create ( topic_1 );
	msg_bus_topic_subscribe ( topic_1, mq1 );

	msg_bus_topic_subscribe_function ( topic_1, recv_cb1, "topic 1 cb" );

	s_topic_msg_t msg;

	while ( 1 )
	{
		if ( rt_mq_recv ( mq1, &msg, sizeof ( msg ), RT_WAITING_FOREVER ) == RT_EOK )
		{
			rt_kprintf ( "recv 1 len:%d %s\r\n", msg.len, msg.p_buf );
		}
	}
}

static void recv_2_entry ( void *para )
{
	mq2 = rt_mq_create ( "mq2", sizeof ( s_topic_msg_t ), 16, RT_IPC_FLAG_FIFO );
	RT_ASSERT ( mq2 );
	msg_bus_topic_subscribe ( topic_2, mq2 );
	msg_bus_topic_subscribe_function ( topic_2, recv_cb2, "topic 2 cb" );
	s_topic_msg_t msg;

	while ( 1 )
	{
		if ( rt_mq_recv ( mq2, &msg, sizeof ( msg ), RT_WAITING_FOREVER ) == RT_EOK )
		{
			rt_kprintf ( "recv 2 len:%d %s\r\n", msg.len, msg.p_buf );
		}
	}
}

static void recv_3_entry ( void *para )
{
	mq3 = rt_mq_create ( "mq3", sizeof ( s_topic_msg_t ), 16, RT_IPC_FLAG_FIFO );
	RT_ASSERT ( mq3 );
	msg_bus_topic_subscribe ( topic_1, mq3 );
	msg_bus_topic_subscribe ( topic_2, mq3 );

	s_topic_msg_t msg;

	while ( 1 )
	{
		if ( rt_mq_recv ( mq3, &msg, sizeof ( msg ), RT_WAITING_FOREVER ) == RT_EOK )
		{
			rt_kprintf ( "recv 3 len:%d %s\r\n", msg.len, msg.p_buf );
		}
	}
}

static void publish_entry ( void *para )
{
	rt_thread_mdelay ( 2000 );
	s_topic_msg_t msg;
	rt_list_t *topic1 = msg_bus_topic_find ( topic_1 );
	rt_list_t *topic2 = msg_bus_topic_find ( topic_2 );

#define SET_MSG(MSG) rt_memset ( msg.p_buf, 0, 64 ); \
	rt_sprintf ( msg.p_buf, "%s", MSG );\
	msg.len = rt_strlen ( msg.p_buf ) \

	while ( 1 )
	{
		rt_thread_mdelay ( 2000 );

		SET_MSG ( "to topic 1 by node" );
		msg_bus_publish_msg_by_node ( topic1, &msg, TOPIC_MSG_SIZE );
		SET_MSG ( "to topic 2 by node" );
		msg_bus_publish_msg_by_node ( topic2, &msg, TOPIC_MSG_SIZE );

		rt_thread_mdelay ( 2000 );
		SET_MSG ( "to topic 1 by node urgent" );
		msg_bus_publish_msg_by_node_urgent ( topic1, &msg, TOPIC_MSG_SIZE );
		SET_MSG ( "to topic 2 by node urgent" );
		msg_bus_publish_msg_by_node_urgent ( topic2, &msg, TOPIC_MSG_SIZE );

		rt_thread_mdelay ( 2000 );
		SET_MSG ( "to topic 1 by name" );
		msg_bus_publish_msg_by_name ( topic_1, &msg, TOPIC_MSG_SIZE );
		SET_MSG ( "to topic 2 by name" );
		msg_bus_publish_msg_by_name ( topic_2, &msg, TOPIC_MSG_SIZE );

		rt_thread_mdelay ( 2000 );
		SET_MSG ( "to topic 1 by name urgent" );
		msg_bus_publish_msg_by_name_urgent ( topic_1, &msg, TOPIC_MSG_SIZE );
		SET_MSG ( "to topic 2 by name urgent" );
		msg_bus_publish_msg_by_name_urgent ( topic_2, &msg, TOPIC_MSG_SIZE );

		rt_thread_mdelay ( 2000 );
		rt_thread_mdelay ( 2000 );
		msg_bus_publish_function_by_node ( topic1 );
		msg_bus_publish_function_by_name ( topic_2 );
	}
}

static int msg_bus_test ( void )
{
	rt_thread_t tid1 = rt_thread_create ( "rcv1", recv_1_entry, 0, 2048, 10, 10 );
	RT_ASSERT ( tid1 != 0 );
	rt_thread_startup ( tid1 );

	rt_thread_t tid2 = rt_thread_create ( "rcv2",  recv_2_entry, 0, 2048, 10, 10 );
	RT_ASSERT ( tid2 != 0 );
	rt_thread_startup ( tid2 );

	rt_thread_t tid3 = rt_thread_create ( "rcv3",  recv_3_entry, 0, 2048, 10, 10 );
	RT_ASSERT ( tid3 != 0 );
	rt_thread_startup ( tid3 );

	rt_thread_t tid4 = rt_thread_create ( "pub",  publish_entry, 0, 2048, 10, 10 );
	RT_ASSERT ( tid4 != 0 );
	rt_thread_startup ( tid4 );

	return 0;
}

/* 导出到初始化列表 */
INIT_APP_EXPORT ( msg_bus_test );

/* 导出到 MSH 命令列表 */
MSH_CMD_EXPORT ( do_topic_enable, set enable );
MSH_CMD_EXPORT ( do_topic_unsub, unsub );
MSH_CMD_EXPORT ( do_topic_sub, sub );
